package com.wpre.vampiremq.mqserver.datacenter;

import com.wpre.vampiremq.common.MQException;
import com.wpre.vampiremq.mqserver.core.Binding;
import com.wpre.vampiremq.mqserver.core.Exchange;
import com.wpre.vampiremq.mqserver.core.MSGQueue;
import com.wpre.vampiremq.mqserver.core.Message;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created with IntelliJ IDEA.
 * Description:管理内存中的数据
 * User: LianBao
 * Date: 2025-01-15
 * Time: 21:37
 */
@Slf4j
public class MemoryDataCenter {
    //存储交换机,key 是 exchangeName,value 是Exchange对象
    //可能会在多线程环境下使用这个类中的方法,所以使用ConcurrentHashMap,
    private ConcurrentHashMap<String, Exchange> exchangeMap
            = new ConcurrentHashMap<>();

    //存储队列,key是queueName,value是MSG对象
    private ConcurrentHashMap<String, MSGQueue> queueMap
            = new ConcurrentHashMap<>();

    //存储绑定,第一个key是exchangeName,第二个key是queueName
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap
            = new ConcurrentHashMap<>();

    //存储消息,key是消息messageId,value是Message对象
    private ConcurrentHashMap<String, Message> messageMap
            = new ConcurrentHashMap<>();

    //存储队列和消息的关联关系,key是queueName,value是Message链表
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap
            = new ConcurrentHashMap<>();

    //存储未被确认的消息,key1是queueName,key2是messageId
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitACKMap
            = new ConcurrentHashMap<>();


    //插入交换机
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
    }
    //获取交换机
    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }
    //删除交换机
    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
    }
    //插入队列
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
    }
    //获取队列
    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }
    //删除队列
    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
    }
    //插入绑定
    public void inertBinding(Binding binding) {
        //先根据exchangeName查找,如果没有就创建一个新的
        ConcurrentHashMap<String, Binding> bindMap =
                bindingsMap.computeIfAbsent(binding.getExchangeName(),
                        k -> new ConcurrentHashMap<>());//如果bindingsMap中没有key对应的value,就会在bindingsMap中创建出一个ConcurrentHashMap<String, Binding>
        //根据binding对象的exchangeName，从bindingsMap中查找对应的内层ConcurrentHashMap。
        //如果不存在，则创建一个新的ConcurrentHashMap并插入到bindingsMap中。computeIfAbsent方法确保了线程安全地创建和插入内层ConcurrentHashMap。

        synchronized (bindMap) {//对内层ConcurrentHashMap进行同步操作，确保在插入绑定时的线程安全。
            //再根据 queueName查找,查找的Binding如果已经存在就抛出异常
            if (bindMap.get(binding.getQueueName()) != null) {
                throw new MQException("绑定已经存在,不能插入");
            }
            //插入操作
            //将新的binding对象插入到内层ConcurrentHashMap中。
            bindMap.put(binding.getQueueName(), binding);

        }
    }
    //获取绑定
    //1.获取exchangeName 和 queueName 确定唯一的Binding
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }
    //2.获取所有的绑定
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }
    //删除绑定
    public void deleteBinding(Binding binding) {
        ConcurrentHashMap<String, Binding> bindingMap
                = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            //该交换机没有绑定任何队列
            throw new MQException("该交换机没有绑定任何队列");
        }
        //删除
        bindingMap.remove(binding.getQueueName());

    }

    //消息管理
    //1.添加消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        log.info("插入消息成功");
    }
    //2.根据id查询消息
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }
    //3.根据id删除消息
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        log.info("消息已被移除");
    }
    // 发送消息到指定的队列
    public void sendMessage(Message message, MSGQueue queue) {
        //computeIfAbsent是线程安全的
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(),
                k -> new LinkedList<>());
        //从queueMessageMap查询对应的链表,如果查不到就创建一个
        synchronized (messages) {
            messages.add(message);
        }
        //加如总消息
        addMessage(message);
    }

    //从队列中取走消息
    public Message pollMessage(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return null;
        }
        synchronized (messages) {
            //如果没有找到,队列中没有任何消息
            if (messages.isEmpty()) {
                return null;
            }
            //把第一个位置的消息删除(只有在消费消息的时候才需要获取,所以是删除)
            return messages.remove(0);
        }
    }
    //获取指定队列中消息的个数
    public int getMessageCount(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return 0;
        }
        synchronized (messages) {
            //获取的时候也可能有线程安全
            return messages.size();
        }
    }
    // 添加未确认消息
    public void addMessageWaitAck(String queueName, Message message) {

        //根据队列名,找到对应的哈希表
        ConcurrentHashMap<String, Message> stringMessageConcurrentHashMap =
                queueMessageWaitACKMap.computeIfAbsent(queueName,
                        k -> new ConcurrentHashMap<>());
        //添加
        stringMessageConcurrentHashMap.put(message.getMessageId(), message);
    }
    //删除未确认消息(消息已经确认了)
    public void removeMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> stringMessageConcurrentHashMap =
                queueMessageWaitACKMap.get(queueName);
        if (stringMessageConcurrentHashMap == null) {
            return;
        }
        //删除
        stringMessageConcurrentHashMap.remove(messageId);
    }
    //获取指定的未确认消息
    public Message getMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> stringMessageConcurrentHashMap =
                queueMessageWaitACKMap.get(queueName);
        if (stringMessageConcurrentHashMap == null) {
            return null;
        }

        return stringMessageConcurrentHashMap.get(messageId);
    }


    //从硬盘中读取数据,恢复到内存中
    public void recovery(DiskDataCenter dataCenter) throws IOException, ClassNotFoundException {

        //先清空内存中原来的数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        queueMessageMap.clear();
        messageMap.clear();
        //1.恢复交换机
        //查询所有的交换机
        List<Exchange> exchanges = dataCenter.selectAllExchanges();
        //循环插入
        for (Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(), exchange);
        }

        //2.恢复队列
        List<MSGQueue> msgQueues = dataCenter.selectAllQueues();
        for (MSGQueue queue : msgQueues) {
            queueMap.put(queue.getName(), queue);
        }

        //3.恢复绑定
        List<Binding> bindings = dataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            //
            ConcurrentHashMap<String, Binding> bindingConcurrentHashMap =
                    bindingsMap.computeIfAbsent(binding.getExchangeName(),//先根据ExchangeName查出对应的Map,如果没有就创建出来
                            k -> new ConcurrentHashMap<>());
            //上述操作,会在bindingsMap中创建一个ConcurrentHashMap<String, Binding>,也就是bindingConcurrentHashMap
            //同时这个bindingConcurrentHashMap的key会被设置成exchangeName,所以后续只需要将value设置一下就好了
            //设置value,value也是一个ConcurrentHashMap
            bindingConcurrentHashMap.put(binding.getQueueName(), binding);
        }

        //4.恢复消息数据
        //遍历所有的队列,根据每个队列名,获取所有的消息,然后插入
        for (MSGQueue queue : msgQueues) {
            // 根据队列名加载出所有的消息
            LinkedList<Message> messages =
                    dataCenter.loadAllMessageFromQueue(queue.getName());
            //把消息加入到queueMessageMap
            queueMessageMap.put(queue.getName(), messages);
            //再插入到messageMap中
            for (Message message : messages) {
                messageMap.put(message.getMessageId(), message);
            }
        }

        //未确认的消息不需要从硬盘中恢复,毕竟之前硬盘存储的设计时,没有考虑这一点
        //等待确认的过程中,一旦服务重启,未被确认的消息又会变成未被取走的状态

    }



}
